package com.wzz.shop.realtime.etl.process

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializerFeature
import com.wzz.canal.util.ip.IPSeeker
import com.wzz.shop.realtime.etl.`trait`.MQBaseETL
import com.wzz.shop.realtime.etl.bean.{ClickLogBean, ClickLogWideBean}
import com.wzz.shop.realtime.etl.utils.GlobalConfigUtil
import nl.basjes.parse.httpdlog.HttpdLoglineParser
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}

import java.io.File



/**
 *
 * @author ${author}
 * @date 2021-09-17 11:29
 *
 */
class ClickLogETL (env:StreamExecutionEnvironment)extends MQBaseETL(env){
  /**
   * 业务处理接口
   */
  override def process(): Unit = {
    val clickLogDS: DataStream[String] = getKafkaDataStream(GlobalConfigUtil.`input.topic.click_log`)
    clickLogDS.printToErr()
    //实时ETL
    val clickLogwideDataStream: DataStream[ClickLogWideBean] = etl(clickLogDS)
    clickLogwideDataStream.print("点击流日志")
    //将clickLogwideBeanDS转换成json字符串返回，因为kafka中需要传入字符串类型的数据
    val clickLogWideJsonDataStream = clickLogwideDataStream.map(clickLogWideEntityEntity => {
      JSON.toJSONString(clickLogWideEntityEntity, SerializerFeature.DisableCircularReferenceDetect)
    })
    clickLogWideJsonDataStream.printToErr("拉宽后的点击流日志》》》")
    //5:将关联维度表后的数据写入到kafka中，供Druid进行指标分析
    clickLogWideJsonDataStream.addSink(kafkaProducer(GlobalConfigUtil.`output.topic.clicklog`))
  }

  /**
   * 数据实时ETL
   * @param clicklogDS
   * @return
   */
  private  def etl(clicklogDS:DataStream[String] ) ={
    // 1. 基于JSON解析字段
    val clickLogBeanDS = clicklogDS.map(new RichMapFunction[String, ClickLogBean] {
      var parser: HttpdLoglineParser[ClickLogBean] = null

      override def open(parameters: Configuration): Unit = {
        parser = ClickLogBean.createClickLogParser()
      }

      override def map(value: String): ClickLogBean = {
        ClickLogBean(parser, value)
      }
    })
    clickLogBeanDS.map(new RichMapFunction[ClickLogBean,ClickLogWideBean] {
      var ipSeeker:IPSeeker = _

      override def open(parameters: Configuration): Unit = {
        val dataFile:File = getRuntimeContext.getDistributedCache.getFile("qqwry.dat")
        ipSeeker = new IPSeeker(dataFile)
      }
      override def map(value: ClickLogBean): ClickLogWideBean = {
        //拷贝原有属性
        val wideBean = ClickLogWideBean(value)
        val country = ipSeeker.getCountry(wideBean.ip)
        var areaArry = country.split("省")
        if(areaArry.length > 1){
          wideBean.province = areaArry(0) + "省";
          wideBean.city = areaArry(1)
        } else {
          areaArry = country.split("市")
          if(areaArry.length>1){
            wideBean.city = areaArry(0) + "市";
            wideBean.city = areaArry(1)
          }
          else {
            wideBean.province = areaArry(0)
            wideBean.city = ""
          }
        }
        wideBean
      }
    })

  }

}
